#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | cst_ds }}'
name='中心+集散进港件总量、建包量统计'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "扫描日期\t扫描站点编码\t扫描站点名称\t站点标识1是中心0是集散\t件总量\t包件量\t包数量\t循环包数量">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 20 --executor-memory 12g --executor-cores 4 --conf spark.debug.maxToStringFields=1000 --conf spark.sql.shuffle.partitions=800 --conf spark.executor.memoryOverhead=6g --conf spark.driver.maxResultSize=3g -S -e "
with send_base as (
    select date_time                                                                                    --日期
         , waybill_no                                                                                   --运单编号_ef
         , input_site                                                                                   --扫描网点_ef
         , scan_site_code                                                                               --扫描网点编号_ef
         , package_code                                                                                 --所属包号_ef
         , scan_time                                                                                    --扫描时间_ef
         , next_station_code                                                                            --下一站
         , have_pack_flag                                                                               --1有包号,0无包号
         , network_type                                                                                 --扫描网点类型
         , next_network_type                                                                            --下一站网点类型
         , case when (network_type = 2 and next_network_type in (3, 4)) then 1 else 0 end as zx_js_flag --1中心0集散
    from (
             select substr(scan_time, 1, 10)                                                            as date_time      --日期
                  , waybill_no                                                                                            --运单编号_ef
                  , input_site                                                                                            --扫描网点_ef
                  , scan_site_code                                                                                        --扫描网点编号_ef
                  , network_type                                                                                          --扫描网点类型
                  , package_code                                                                                          --所属包号_ef
                  , scan_time                                                                                             --扫描时间_ef
                  , next_station_code                                                                                     --下一站
                  , next_network_type                                                                                     --下一站网点类型
                  , if(package_code is not null, 1, 0)                                                  as have_pack_flag --1有包号,0无包号
                  , row_number() over (partition by waybill_no,scan_site_code order by scan_time desc ) as rn
             from jms_dwd.dwd_tab_barscan_send_base_dt
             where dt  = '{{ execution_date | cst_ds }}'
               and substr(scan_time, 1, 10)  = '{{ execution_date | cst_ds }}'
               and ((network_type = 2 and next_network_type in (3, 4)) or (network_type = 3 and next_network_type = 4))
               and waybill_no is not null
               and substr(waybill_no, 1, 2) = 'JT'
         ) a
    where rn = 1
),
     bagging_base as (
         select waybill_no                    --运单号
              , max(flight_num) as flight_num --芯片号
         from jms_dwd.dwd_tab_barscan_bagging_base_dt
         where dt between date_add('{{ execution_date | cst_ds }}', -19) and date_add('{{ execution_date | cst_ds }}', 1)
           and (waybill_no like 'B%' or waybill_no like '4%')
           and waybill_no is not null
         group by waybill_no
     ),
     join_table as (
         select t1.date_time                                           --日期
              , t1.waybill_no                                          --运单编号_ef
              , t1.input_site                                          --扫描网点_ef
              , t1.scan_site_code                                      --扫描网点编号_ef
              , t1.package_code                                        --所属包号_ef
              , t1.scan_time                                           --扫描时间_ef
              , t1.next_station_code                                   --下一站
              , t1.have_pack_flag                                      --1有包号,0无包号
              , t1.network_type                                        --扫描网点类型
              , t1.next_network_type                                   --下一站网点类型
              , t1.zx_js_flag
              , t4.flight_num                       as flight_num      --芯片号
              , if(t4.flight_num is not null, 1, 0) as flight_num_flag --1有芯片号,0没有芯片号
              , t1.date_time                        as dt
         from send_base t1
                  left join bagging_base t4
                            on t1.package_code = t4.waybill_no
     ),
     zx_js_loop_pack as (--(末端中心)统计站点对应的不重复包数,以及包数中的循环袋的包数
         select date_time
              , scan_site_code
              , zx_js_flag
              , count(package_code)                   as pack_num --包数
              , sum(if(flight_num is not null, 1, 0)) as loop_num --循环袋数
         from (
                  select date_time
                       , scan_site_code --扫描网点编号
                       , package_code   --包号
                       , zx_js_flag
                       , max(flight_num) as flight_num
                  from join_table
                  where package_code is not null
                  group by date_time
                         , scan_site_code --扫描网点编号
                         , package_code   --包号
                         , zx_js_flag
              ) a
         group by date_time
                , scan_site_code
                , zx_js_flag
     ),
     zx_js_result as (--中心的数据
         select a.date_time                as date_time        --日期
              , a.scan_site_code           as scan_site_code   --扫描网点编号
              , a.input_site               as input_site       --扫描网点
              , a.zx_js_flag               as zx_js_flag       --1中心0集散
              , nvl(a.waybill_total, 0)    as waybill_total    --进港总量
              , nvl(a.waybill_pack_num, 0) as waybill_pack_num --包内件
              , nvl(b.pack_num, 0)         as pack_num         --包数
              , nvl(b.loop_num, 0)         as loop_num         --循环袋数
         from (
                  select date_time
                       , scan_site_code                          --扫描网点编号
                       , input_site                              --扫描网点
                       , zx_js_flag
                       , count(waybill_no)   as waybill_total    --进港总量
                       , sum(have_pack_flag) as waybill_pack_num --包内件
                  from join_table
                  group by date_time
                         , scan_site_code --扫描网点编号
                         , input_site     --扫描网点
                         , zx_js_flag
              ) a
                  left join zx_js_loop_pack b
                            on a.scan_site_code = b.scan_site_code
                                and a.date_time = b.date_time
                                and a.zx_js_flag = b.zx_js_flag
     )
select date_time        --日期
     , scan_site_code   --扫描网点编号
     , input_site       --扫描网点
     , zx_js_flag       --1中心0集散
     , waybill_total    --进港总量
     , waybill_pack_num --包内件
     , pack_num         --包数
     , loop_num         --循环袋数
from zx_js_result
;
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt